package com.zhang.sql2;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @title:
 * @author: zhang
 * @date: 2022/3/23 11:04
 */
public class TopNExample {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //读取数据源
        //TODO 1 创建表 (连接器表) 在创建表的ddl中直接定义时间属性
        String create = "create table clickTable (" +
                " name STRING," +
                " url STRING," +
                " ts BIGINT," +
                " et AS TO_TIMESTAMP( FROM_UNIXTIME(ts/1000) )," +
                " WATERMARK FOR et AS et - INTERVAL '1' SECOND " +
                " ) WITH ( " +
                " 'connector' = 'filesystem'," +
                " 'path' = 'input/click.txt'," +
                " 'format' = 'csv'" +
                " )";
        tableEnv.executeSql(create);

        //TODO 普通Top N 选取所有用户中浏览量最大的2个
        Table table = tableEnv.sqlQuery("" +
                " select name," +
                " count(*) cnt" +
                " from clickTable" +
                " group by name");
        //tableEnv.createTemporaryView("countTable", table);

        String sql = " select * from ( " +
                " select *," +
                " ROW_NUMBER() OVER( " +
                " ORDER BY cnt desc ) as rk from countTable ) " +
                " where rk <= 2";
       // Table table1 = tableEnv.sqlQuery(sql);
        //tableEnv.toChangelogStream(table1).print();




        String innerSql =   " select name," +
                " count(url) as cnt ," +
                " window_start, " +
                " window_end " +
                " FROM TABLE(" +
                " TUMBLE( TABLE clickTable, " +
                " DESCRIPTOR(et), " +
                " INTERVAL '5' SECOND) )" +
                " group by name,window_start,window_end";

        String sql2 = " select * from ( " +
                " select *," +
                " ROW_NUMBER() OVER( " +
                " PARTITION BY window_start,window_end " +
                " ORDER BY cnt desc ) as rk from ( " + innerSql + " ) " +
                " ) where rk <= 2";
        Table widowTopN = tableEnv.sqlQuery(sql2);
        tableEnv.toDataStream(widowTopN).print();

        env.execute();
    }
}
